package cn.itcast.streaming.task;

import cn.itcast.streaming.bean.ElectricFenceModel;
import cn.itcast.streaming.bean.ElectricFenceRuleTmp;
import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.function.flatmap.ElectricFenceModelFunction;
import cn.itcast.streaming.function.flatmap.ElectricFenceRulesFunction;
import cn.itcast.streaming.function.watermark.ElectricFenceWaterMark;
import cn.itcast.streaming.function.window.ElectricFenceWindowFunction;
import cn.itcast.streaming.sink.ELectricFenceMysqlSink;
import cn.itcast.streaming.source.MysqlElectricFenceResultSource;
import cn.itcast.streaming.source.MysqlElectricFenceSource;
import cn.itcast.streaming.utils.JsonParseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.HashMap;

//电子围栏实时业务分析，将消费的数据处理后与电子围栏规则数据进行逻辑处理，
//如果是电子围栏中的车辆，将进入电子围栏时间和出围栏时间写入到mysql数据库中
public class ElectricFenceTask extends BaseTask{
    public static void main(String[] args) throws Exception {
        /**
         * 实现步骤
         *  1）初始化flink环境
         *  2）接入kafka数据源消费数据
         *  3）将json字符串解析为javabean对象
         *  4）过滤出正常数据
         *  5）获取到电子围栏规则数据以及电子围栏关联的车辆数据，并进行广播
         *  6）将原始数据与电子围栏规则广播流数据进行合并connect
         *  7）对合并后的数据分组后用90s滚动窗口进行自定义窗口函数的开发
         *  8）读取电子围栏分析结果表数据进行广播
         *  9）对第七步和第八步的数据结果进行合并connect
         *  10)对第九步的结果进行滚动窗口操作，应用自定义函数，增加UUID和inMysql字段
         *  11）将分析好的电子围栏结果数据实时写入mysql中
         *  12）运行作业
         */


        //todo 1）初始化flink环境
        StreamExecutionEnvironment env = getEnv("ElectricFenceTask");
        env.setParallelism(1);
        //todo 2）接入kafka数据源消费数据
        DataStream<String> kafkaStream = createKafkaStream(SimpleStringSchema.class);
        //todo 3）将json字符串解析为javabean对象
        SingleOutputStreamOperator<ItcastDataObj> itcastDataObjStream = kafkaStream
                .filter(obj -> StringUtils.isNotEmpty(obj)).map(JsonParseUtil::parseJsonToObject);
        //todo 4）过滤出正常数据
        SingleOutputStreamOperator<ItcastDataObj> srcItcastObjStream = itcastDataObjStream
                .filter(ojb -> StringUtils.isEmpty(ojb.getErrorData()));
        //todo 5）获取到电子围栏规则数据以及电子围栏关联的车辆数据，并进行广播
        DataStream<HashMap<String, ElectricFenceRuleTmp>> electricFenceVinsStream = env
                .addSource(new MysqlElectricFenceSource()).broadcast();
        //todo 6）将原始数据与电子围栏规则广播流数据进行合并connect
        ConnectedStreams<ItcastDataObj, HashMap<String, ElectricFenceRuleTmp>>
                electricFenceVinsConnectStream = srcItcastObjStream.connect(electricFenceVinsStream);
        //将电子围栏规则流数据应用到原始数据流中（拉宽）
        SingleOutputStreamOperator<ElectricFenceModel> electricFenceModel = electricFenceVinsConnectStream
                .flatMap(new ElectricFenceRulesFunction());

        //electricFenceModel.print("原始数据与电子围栏规则广播流进行拉宽后的结果>>>");
        //todo 7）对合并后的数据分组后用90s滚动窗口进行自定义窗口函数的开发
        SingleOutputStreamOperator<ElectricFenceModel> electricFenceModelDataStream =
                electricFenceModel.assignTimestampsAndWatermarks(new ElectricFenceWaterMark())
                .keyBy(ElectricFenceModel::getVin)
                .window(TumblingEventTimeWindows.of(Time.seconds(90)))
                .apply(new ElectricFenceWindowFunction());

        //electricFenceModelDataStream.printToErr("电子围栏拉宽后的结果>>>");
        //todo 8）读取电子围栏分析结果表数据进行广播
        DataStream<HashMap<String, Long>> electricFencemodelResultData = env.addSource(new MysqlElectricFenceResultSource()).broadcast();
        //electricFencemodelResultData.print("电子围栏分析结果表>>>");
        //todo 9）对第七步和第八步的数据结果进行合并connect
        ConnectedStreams<ElectricFenceModel, HashMap<String, Long>> connectStream = electricFenceModelDataStream.connect(electricFencemodelResultData);
        //todo 10)对第九步的结果进行滚动窗口操作，应用自定义函数，增加UUID和inMysql字段
        SingleOutputStreamOperator<ElectricFenceModel> resultDataStream = connectStream
                .flatMap(new ElectricFenceModelFunction());
        //resultDataStream.print("最终结果数据：");
        //todo 11）将分析好的电子围栏结果数据实时写入mysql中
        resultDataStream.addSink(new ELectricFenceMysqlSink());
        //todo 12）运行作业
        env.execute();

    }

}
